package cn.bigdata.sparkstraming.job

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 *
 * 将sample.log的数据发送到Kafka中，经过Spark Streaming处理，将数据格式变为以下形式：
 *
 * commandid | houseid | gathertime | srcip | destip |srcport| destport | domainname | proxytype | proxyip | proxytype | title | content | url | logid
 *
 * 在发送到kafka的另一个队列中
 * 要求：
 * 1、sample.log => 读文件，将数据发送到kafka队列中
 * 2、从kafka队列中获取数据（0.10 接口不管理offset），变更数据格式
 * 3、处理后的数据在发送到kafka另一个队列中
 */
object Case1 {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf = new SparkConf().setAppName(this.getClass.getName.init).setMaster("local")

    // 每5秒一个批次处理数据
    val ssc = new StreamingContext(conf, Seconds(5))

    val topic = "sample-log"
    val topic2 = "sample-log-2"
    val groupId = s"${topic}-groupid"

    // 这里使用本地文件
    val lines = ssc.textFileStream("data/log/")
    lines.map(_.trim.split("<<<!>>>")).map(x => x.mkString("")).map(_.split(","))
      .map(x => {
        val log = x.mkString(" | ")
        SampleLogSender.sendLog(topic, log)
        log
      }).count().print()

    // 消费 sample-log 的数据， 并写入到另一个topic sample-log-2
    val kafkaDStream = KafkaUtils.createDirectStream(ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array(topic), getKafkaParams(groupId))
    )
    kafkaDStream.map(x => {
      SampleLogSender.sendLog(topic2, x.value())
      x.value()
    }).count().print()

    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaParams(groupId: String): Map[String, Object] = {
    Map(
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean),
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "centos2:9092,centos3:9092,centos4:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }
}
